package net.wicp.tams.common.binlog.alone.checkpoint;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Conf;
import net.wicp.tams.common.apiext.CollectionUtil;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.apiext.jdbc.JdbcAssit;
import net.wicp.tams.common.apiext.jdbc.JdbcConnection;
import net.wicp.tams.common.binlog.alone.ListenerConf.ColHis;
import net.wicp.tams.common.binlog.alone.ListenerConf.ColHis.Builder;
import net.wicp.tams.common.binlog.alone.ListenerConf.ConnConf;
import net.wicp.tams.common.binlog.alone.ListenerConf.Position;
import net.wicp.tams.common.binlog.alone.normalize.ISaveCheckPoint;
import net.wicp.tams.common.constant.dic.YesOrNo;
import net.wicp.tams.common.exception.ExceptAll;
import net.wicp.tams.common.exception.ProjectExceptionRuntime;

@Slf4j
public class CheckPointMysql implements ISaveCheckPoint {
	// private static Server server;
	private Connection connection = null;

	private Connection connectionLock = null;

	private String url = "";

	private ConnConf connConf;

	private String lockTableName = "";

	@Override
	public void init(ConnConf.Builder connConfBuilder) {
		try {
			Map<String, String> pre = Conf.getPre("common.binlog.alone.chk.mysql", true);
			boolean isSameListener = StringUtil.isNull(pre.get("ip")) ? true : false;// 是不是与监听同一个数据库
			url = String.format("jdbc:mysql://%s:%s/%s?autoReconnect=true&useUnicode=true&characterEncoding=utf-8",
					isSameListener ? Conf.get("common.binlog.alone.conf.default.ip") : pre.get("ip"),
					isSameListener ? Conf.get("common.binlog.alone.conf.default.port") : pre.get("port"),
					pre.get("defaultdb"));
			connection = JdbcConnection.getConnectionMyql(url,
					isSameListener ? Conf.get("common.binlog.alone.conf.default.user") : pre.get("user"),
					isSameListener ? Conf.get("common.binlog.alone.conf.default.password") : pre.get("password"));
			Statement stmt = connection.createStatement();
			int executeUpdate = stmt.executeUpdate(
					"CREATE TABLE IF NOT EXISTS `position`  (`gtids` varchar(255) NOT NULL,`fileName` varchar(255) NULL,`pos` bigint NULL,`masterServerId` bigint NULL,`time` bigint NULL,`timeStr` varchar(255) NULL,`serverIp` varchar(255) NOT NULL,`clintId` varchar(255) NOT NULL,PRIMARY KEY (`gtids`))");

			int executeUpdate2 = stmt.executeUpdate(
					"CREATE TABLE IF NOT EXISTS `colhis`  (`db` varchar(255) NOT NULL,`tb` varchar(255) NOT NULL,`time` bigint NOT NULL,`timeStr` varchar(255)  NULL,`serverIp` varchar(255) NOT NULL,`cols` varchar(5000) NOT NULL,`coltypes` varchar(3000) NOT NULL,PRIMARY KEY (`db`,`tb`, `time`))");

			int executeUpdate3 = 0;
			if (StringUtil.isNotNull(connConfBuilder.getGroupId())) {
				connectionLock = JdbcConnection.getConnectionMyql(url,
						isSameListener ? Conf.get("common.binlog.alone.conf.default.user") : pre.get("user"),
						isSameListener ? Conf.get("common.binlog.alone.conf.default.password") : pre.get("password"));
				lockTableName = String.format("t_lock_%s_%s", connConfBuilder.getGroupId(), connConfBuilder.getIp())
						.replaceAll("-", "_").replaceAll("\\.", "_");
				executeUpdate3 = stmt.executeUpdate(
						"CREATE TABLE IF NOT EXISTS `" + lockTableName + "`  (`id` int NOT NULL,PRIMARY KEY (`id`))");
			}

			stmt.close();
			if (executeUpdate != 0 || executeUpdate2 != 0
					|| (StringUtil.isNotNull(lockTableName) && executeUpdate3 != 0)) {
				throw new ProjectExceptionRuntime(ExceptAll.jdbc_exec_fail, "创建table失败");
			}
			this.connConf = connConfBuilder.build();
			log.info("execute=" + executeUpdate);
		} catch (Exception e) {
			log.error("初始化失败", e);
			throw new RuntimeException("初始化失败", e);
		}
	}

	@Override
	public void shutdown() {
		try {
			if (connection != null && !connection.isClosed()) {
				connection.close();
			}
		} catch (Exception e) {
			log.error("关闭连接失败", e);
		}
	}

	private PreparedStatement pointPrep;

	@Override
	public void savePoint(Position pos) {
		// replace在1.8以上才支持
		this.pointPrep = checkStmt(
				"replace into position (gtids,filename,pos,masterserverid,time,timeStr,serverIp,clintId) values(?,?,?,?,?,?,?,?)",
				this.pointPrep);
		try {
			JdbcAssit.setPreParam(pointPrep, pos.getGtids(), pos.getFileName(), pos.getPos(), pos.getMasterServerId(),
					pos.getTime(), pos.getTimeStr(), pos.getServerIp(), pos.getClintId());
			pointPrep.executeUpdate();
			log.info("保存实例:{} 位点{}成功", pos.getServerIp(), pos.getGtids());
		} catch (Exception e) {
			log.error("保存位点失败,实例:" + pos.getServerIp() + " 位点:" + pos.getGtids(), e);
		}
	}

	private PreparedStatement checkStmt(String sql, PreparedStatement stmt) {
		PreparedStatement returnobj = stmt;
		while (true) {
			try {
				if (this.connection == null || this.connection.isClosed()) {
					// 先关闭旧的stmt;
					if (stmt != null && stmt.isClosed()) {
						stmt.close();
					}
					this.connection = DriverManager.getConnection(url, "sa", "");
				}
				if (stmt == null) {
					returnobj = connection.prepareStatement(sql);
				}
				break;
			} catch (Exception e) {
				log.error("数据库连接不上或创建stmt失败", e);
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e1) {
				}
			}
		}
		return returnobj;
	}

	private PreparedStatement colSaveColsPre;

	@Override
	public void saveColName(ColHis colHis) {
		this.colSaveColsPre = checkStmt(
				"replace into colhis (db,tb,time,timeStr,serverIp,cols,coltypes) values(?,?,?,?,?,?,?)",
				this.colSaveColsPre);
		try {
			String cols = CollectionUtil.listJoin(colHis.getColsList(), ",");
			String colstype = CollectionUtil.listJoin(colHis.getColTypesList(), ",");
			JdbcAssit.setPreParam(this.colSaveColsPre, colHis.getDb(), colHis.getTb(), colHis.getTime(),
					colHis.getTimeStr(), colHis.getServerIp(), cols, colstype);
			colSaveColsPre.executeUpdate();
			log.info("保存字段名成功,db{},tb{},time{}", colHis.getDb(), colHis.getTb(), colHis.getTime());
		} catch (Exception e) {
			log.error("保存位点失败", e);
		}
	}

	private PreparedStatement queryPointPre;

	@Override
	public Position findPoint(long time) {
		this.queryPointPre = checkStmt(
				"select * from position  where serverIp=? and clintId=? and time<=?   order by time desc limit 0,1",
				this.queryPointPre);
		Position retojb = null;
		try {
			JdbcAssit.setPreParam(this.queryPointPre, connConf.getIp(), connConf.getClientId(), time);
			ResultSet executeQuery = queryPointPre.executeQuery();
			if (executeQuery.next()) {
				Position.Builder builder = Position.newBuilder();
				builder.setFileName(executeQuery.getString("fileName"));
				builder.setGtids(executeQuery.getString("gtids"));
				builder.setPos(executeQuery.getLong("pos"));
				builder.setMasterServerId(executeQuery.getLong("masterServerId"));
				builder.setTime(executeQuery.getLong("time"));
				builder.setTimeStr(executeQuery.getString("timeStr"));
				builder.setServerIp(executeQuery.getString("serverIp"));
				builder.setClintId(executeQuery.getInt("clintId"));
				retojb = builder.build();
			}
		} catch (SQLException e) {
			log.error("查询位点失败", e);
		}
		return retojb;
	}

	private PreparedStatement queryColsPre;

	@Override
	public List<ColHis> findColsList(String db, String tb) {
		this.queryColsPre = checkStmt("select * from colhis  where db=? and tb=? and serverIp=? order by time desc",
				this.queryColsPre);
		List<ColHis> retlist = new ArrayList<>();
		try {
			JdbcAssit.setPreParam(this.queryColsPre, db, tb, connConf.getIp());
			ResultSet resultSet = queryColsPre.executeQuery();
			while (resultSet.next()) {
				Builder tempBuilder = ColHis.newBuilder();
				tempBuilder.setDb(db);
				tempBuilder.setTb(tb);
				tempBuilder.setTime(resultSet.getLong("time"));
				tempBuilder.setTimeStr(resultSet.getString("timeStr"));
				tempBuilder.setServerIp(resultSet.getString("serverIp"));
				String[] colsAry = resultSet.getString("cols").split(",");
				tempBuilder.addAllCols(Arrays.asList(colsAry));
				String[] coltypesAry = resultSet.getString("coltypes").split(",");
				tempBuilder.addAllColTypes(Arrays.asList(coltypesAry));
				retlist.add(tempBuilder.build());
			}
			resultSet.close();
		} catch (Exception e) {
			log.error("查colname失败", e);
		}
		return retlist;
	}

	@Override
	public List<ColHis> findColsAll() {
		List<ColHis> retlist = new ArrayList<>();
		try {
			ResultSet resultSet = JdbcAssit.querySql(connection,
					"select * from colhis where serverIp='" + connConf.getIp() + "' order by time desc");
			while (resultSet.next()) {
				Builder tempBuilder = ColHis.newBuilder();
				tempBuilder.setDb(resultSet.getString("db"));
				tempBuilder.setTb(resultSet.getString("tb"));
				tempBuilder.setTime(resultSet.getLong("time"));
				tempBuilder.setTimeStr(resultSet.getString("timeStr"));
				tempBuilder.setServerIp(resultSet.getString("serverIp"));
				String[] colsAry = resultSet.getString("cols").split(",");
				tempBuilder.addAllCols(Arrays.asList(colsAry));
				String[] coltypesAry = resultSet.getString("coltypes").split(",");
				tempBuilder.addAllColTypes(Arrays.asList(coltypesAry));
				retlist.add(tempBuilder.build());
			}
			resultSet.close();
		} catch (Exception e) {
			log.error("查colname all失败", e);
		}
		return retlist;
	}

	// 获取分布式锁
	// 行级锁需要查询在唯一索引上，但由于InnoDB引擎优化作用，可能由行级锁变为表级锁，所以干脆用表级锁来处理
	@Override
	public YesOrNo acquireLock() {
		if (StringUtil.isNull(lockTableName)) {
			return YesOrNo.yes;
		}
		while (true) {
			try {
				connectionLock.setAutoCommit(false);
				Statement stmt = connectionLock.createStatement();
				stmt.setQueryTimeout(60);// 设置超时时间 10秒会报Statement cancelled due to timeout or client request错误
				stmt.executeUpdate("replace into " + lockTableName + " value(1)");
				ResultSet result = stmt.executeQuery("select * from " + lockTableName + " where id=1 for update");
				if (result != null && result.next() && result.getInt(1) == 1) {
					// connection.setAutoCommit(true);
					return YesOrNo.yes;
				}
				Thread.sleep(1000);
			} catch (Exception e) {
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e1) {
				}
				log.info("acquireLock error,No distributed locks were obtained!");
			}
		}
	}

	// 释放分布式锁
	@Override
	public void releaseLock() {
		if (StringUtil.isNull(lockTableName)) {
			return;
		}
		try {
			connectionLock.commit();
		} catch (SQLException e) {
			log.error("releaseLock error!", e);
		}
	}

}
